package book.data_clean;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.log4j.BasicConfigurator;

import java.io.IOException;

public class CleanDriver {
    public static void main(String[] args) {
        //日志
        BasicConfigurator.configure();

        try {
            //配置环境
            Configuration conf = new Configuration();
            conf.set("fs.default", "hdfs://192.168.0.155:9000/");
            Job job = Job.getInstance(conf);
            //设置driver、mapper和driver类
            job.setJarByClass(CleanDriver.class);
            job.setMapperClass(CleanMapper.class);
            job.setReducerClass(ClearReducer.class);

            //设置map输出的key和value的数据类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);

            //设置map输出的key和value的数据类型
            job.setOutputKeyClass(Text.class);
            job.setOutputKeyClass(NullWritable.class);

            //文件的读取路径
            FileInputFormat.setInputPaths(job, "/MR/input");

            //结果的数据路径
            Path path = new Path("/MR/out");
            //如果路径存在那么就删除
            if (path.getFileSystem(conf).exists(path)) {
                path.getFileSystem(conf).delete(path, true);
            }
            FileOutputFormat.setOutputPath(job, path);
            System.out.println(job.waitForCompletion(true) ? 0 : 1);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }
}
